/*
Copyright 2010 Bryce Alcock

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.com

 */
package com.ecocrypt.proxy.forwarder;

import com.ecocrypt.proxy.forwarder.remote.Command;
import com.ecocrypt.proxy.forwarder.remote.ForwarderDynamicStateChangedPayload;
import com.ecocrypt.proxy.forwarder.transfer.ForwarderDataChangeNotification;
import com.ecocrypt.proxy.forwarder.transfer.ForwarderListener;
import java.io.IOException;
import java.io.Serializable;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.StringTokenizer;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * This Forwarder represents a socket server that will proxy all socket communication
 * comming into this localhost port to the remoteMachine and port.
 * <br>
 * This server works by listening for new socket requests (incomming socket requests).
 * When one arrives, it will start up 4 threads to help service this requests, future
 * version of this could be enhanced to use thread service pools and tasks. The 4  threads are
 * <br>
 * <ul>
 * <li>Client Socket Input Thread - continuously pull data off the client requests input stream and store in client input buffer (CIB)</li>
 * <li>Client Socket Output Thread - continuously writes data back to the client output stream from the EOB</li>
 * <li>Engine Socket Output Thread - continuously writes to this stream from the CIB</li>
 * <li>Engine Socket Input Thread - continuously pull data off the engine input socket and store in engine output buffer (EOB)</li>
 * </ul>
 * This System has 2 main buffers.
 * <ul>
 * <li>CIB - the Client Input Buffer holds all data read from the client request
 * input socket that has not yet been written to the Engine Socket Output Thread.</li>
 * <li>EOB - the Engine Output Buffer holds all of the data read from the engine
 * that has not yet been processed by the,
 * client socket output thread.
 * </li>
 * </ul>
 *
 * <br>
 * This class is final because
 * it would be dangerous to extend this.
 * People with the Source code can modify the code and compile.
 * or use aggregation/decorator design patterns to extend.
 *
 * @author Bryce.Alcock
 * @since 1.0
 */
public class Forwarder implements Runnable, Serializable, ForwarderDataChangeNotification {

    transient Timer t;// = new Timer("[Forwarder Timer]" + this.toString());

    // The Request SerializationBuffer is a buffer to store data prior to getting serialized to disk.
    transient RequestSerializationBuffer rsb;
    //  Used for notification of new items.
    transient protected List<ForwarderListener> listenerList = new ArrayList<ForwarderListener>();
    //-------------------------------------------------------------------------
    //
    //   Data associated with the immutable characteristics of this forwarder.
    //   The only way to affect these values is at object creation time.
    //
    //--------------------------------------------------------------------------
    transient private Thread thisSocketServer;
    final int localPort;
    final String remoteMachine;
    final int remotePort;
    //--------------------------------------------------------------------------
    //
    //  Thse are the dynamic state of this forwarder.
    //  requets - represents all of the current in process requests being serviced.
    //
    //  currentRequestId indicates the total number of incomming socket requests 
    //                   that have been attempted to be serviced.
    //--------------------------------------------------------------------------
    private transient final HashMap<ClientServiceRequest, Integer> requests = new HashMap<ClientServiceRequest, Integer>();
    public transient AtomicInteger currentRequestId = new AtomicInteger(0);
    public transient AtomicInteger errorCount = new AtomicInteger(0);
    public transient AtomicInteger activeCount = new AtomicInteger(0);
    //--------------------------------------------------------------------------
    //  These are all public, so that Engine Socket and ClientServiceRequest
    //  can increment without method call overhead.
    //  as well as allowing out of package view of data.
    public transient AtomicLong totalBytesToEngine = new AtomicLong(0);
    public transient AtomicLong totalBytesFromEngine = new AtomicLong(0);
    public transient AtomicLong totalBytesToClient = new AtomicLong(0);
    public transient AtomicLong totalBytesFromClient = new AtomicLong(0);

    /**
     * Package Private Constructor used to allow this otherwise "final" class
     * to be overridden.  This is being overridden to allow for a "Remotely" accessed
     * view of this forwarder.
     */
    Forwarder(String forwardDefinition, boolean b) {

        if (b) {
            String s = forwardDefinition;
            StringTokenizer st = new StringTokenizer(s, ":");

            if (st.countTokens() != 3) {
                ForwardProxy.help();
                throw new IllegalArgumentException(s + ", must have only 3 arguments when tokenized with :, it has ->" + st.countTokens());
            }

            try {
                localPort = Integer.parseInt(st.nextToken());
                remoteMachine = st.nextToken();
                remotePort = Integer.parseInt(st.nextToken());
            } catch (Exception e) {
                throw new IllegalArgumentException(s + "Failed to initialize with the underlying excption message of -> " + e.getMessage(), e);
            }
        } else {
            localPort = -1;
            remoteMachine = "unassigned";
            remotePort = -1;

        }

    }

    /**
     * This will create a new Port forwarder, who purpose is to proxy incomming requests
     * to the local port to remoteMachine:remotePort.  This constructor will also create a new
     * Thread with a thread name representing this port forard defintion in the format:
     * <br>
     * [PortForward] SocketServer: localPort:RemoteMachine:remotePort
     * </br>
     * <b>The Thread Created will not be started until it's start method is called.</b>
     *
     * @param forwardDefinition - this is the localPort:remoteMachine:remotePort
     * String defintion of the immutable charactersistics of this forwarder.
     * @throws IllgalArgumentException if the string is not properly tokenized,
     * or if localPort is not an int, or if remotePort is not an int.
     */
    public Forwarder(String forwardDefinition) {

        String s = forwardDefinition;
        StringTokenizer st = new StringTokenizer(s, ":");

        if (st.countTokens() != 3) {
            ForwardProxy.help();
            throw new IllegalArgumentException(s + ", must have only 3 arguments when tokenized with :, it has ->" + st.countTokens());
        }

        try {
            localPort = Integer.parseInt(st.nextToken());
            remoteMachine = st.nextToken();
            remotePort = Integer.parseInt(st.nextToken());
        } catch (Exception e) {
            throw new IllegalArgumentException(s + "Failed to initialize with the underlying excption message of -> " + e.getMessage(), e);
        }

        thisSocketServer = new Thread(this, "[PortForward] SocketServer: " + toString());

        rsb = new RequestSerializationBuffer(this);
        t = new Timer("[Forwarder Timer]" + this.toString());
        TimerTask simpleRecords = new TimerTask() {

            @Override
            public void run() {
                try {
                    broadcastCurrentStatus();
                } catch (Throwable t) {
                    t.printStackTrace(System.out);
                }
            }
        };
        if(ForwardProxy.allowRemoteConsole){
            t.scheduleAtFixedRate(simpleRecords, 10000, 1000 * 5);
        }
    }

    /**
     * This run method will Start a Server Listener Socket to accept incoming requests
     * on the local port which was set as part of the immutable characteristics of this
     * class.
     * <br>
     * When an incoming request is received, and the Duplex Socket has been accepted,
     * (accepted means a full duplex socket has been established syn/ack/acknak ...)
     * then the following happens:
     * <ol>
     * <li>the currentRequestId is assign as a serviceId and incremented</li>
     * <li>ClientServiceRequest instance is created to service this request.</li>
     * <li>The requests List is modified to include the newly created ClientServiceRequest</li>
     * <li>the ClientServiceRequest Threads are started.</li>
     * </ol>
     *
     * <br><b> Once the LISTENER starts, No EXCEPTIONS will interrupt this thread.</b></br>
     */
    public void run() {

        ServerSocket ss;
        try {
            ss = new ServerSocket(localPort);
        } catch (IOException ex) {
            ex.printStackTrace(System.out);
            throw new RuntimeException("Can't start the Forwarder Call get next exception to see why-> " + ex.getMessage(), ex);
        }


        while (true) {
            try {
                Socket requestSocket = ss.accept();
                //  Now we have to start 2 threads.
                //  One to read from the socket,
                //  and one to right back to the socket.

                Integer serviceId = new Integer(currentRequestId.incrementAndGet());
                ClientServiceRequest csr = new ClientServiceRequest(this, requestSocket, serviceId);
                synchronized (requests) {
                    requests.put(csr, serviceId);
                    activeCount.incrementAndGet();
                }

                csr.startServiceThreads();
                Command c = new Command(Command.ADD_SERVICE_REQUEST, csr);
                ForwardProxy.addServiceMessage(c);

                //  @activeRequests Changed.
                //  For remotely connected machines, we will
                //  have additional complexity of synchronizing this data.




            } catch (Throwable ex) {

                System.out.println("Error Accepting a Socket for " + localPort + ":" + remoteMachine + ":" + remotePort + " Total Error count->" + errorCount.incrementAndGet());
                ex.printStackTrace(System.out);
            }
        }
    }

    // Remove Finished Service Requests.
    public void removeServiceRequest(ClientServiceRequest csr) {
        synchronized (requests) {
            Integer ret = requests.remove(csr);
            if (ret != null) {
                activeCount.decrementAndGet();
                rsb.add(csr);
                for (ForwarderListener forwarderListener : listenerList) {
                    forwarderListener.connectionRemoved(this, csr);
                }
                Command c = new Command(Command.REMOVE_SERVICE_REQUEST, csr);
                ForwardProxy.addServiceMessage(c);
            }
        }
    }

    @Override
    public String toString() {
        // This string should be initialized only the first time this is called.
        final String instanceString = String.valueOf(localPort) + ":" + remoteMachine + ":" + String.valueOf(remotePort);
        return instanceString;
    }

    //@packagePrivate only other members of this package should have visibility to this thread object.
    Thread getThread() {
        return thisSocketServer;
    }

    /**
     * Quick Summary Stats, this method call should be very lightweight as it is
     * only looking at primitives and does very little string manipulation.
     * @return A quick summary of the state of this forwarder.
     */
    public String summaryStats() {
        // String Formatter using relative indexing, and auto boxing of AtomicIntegers to integer formats.
        final String formatString = "[Forwarder Summary] {" + this.toString() + "} Active -> %d "
                + "Forwarder Service Create Errors -> %d "
                + " Total Reqs -> %d ";
        return String.format(formatString, activeCount.get(), errorCount.get(), currentRequestId.get());
    }

    /**
     * A detail view of this Forwarder.
     * This will print 1 line item for each active Cleint Request.
     *
     * @return
     */
    public String detailStats() {

        final String detailStatsHead = "[[Forwarder Details]] %s\n";
        final String row = "\tRequestId %d -> %s \n";

        StringBuilder sb = new StringBuilder();
        sb.append(String.format(detailStatsHead, summaryStats()));
        synchronized (requests) {
            for (ClientServiceRequest request : requests.keySet()) {
                Integer reqId = requests.get(request);
                sb.append(String.format(row, reqId, request));
            }
        }
        return sb.toString();
    }

    public ArrayList<ClientServiceRequest> getCurrentRequests() {
        ArrayList<ClientServiceRequest> reqs = new ArrayList<ClientServiceRequest>();

        synchronized (requests) {
            for (ClientServiceRequest clientServiceRequest : requests.keySet()) {
                reqs.add(clientServiceRequest);
            }
        }
        return reqs;
    }

    //@StateChange
    private void broadcastCurrentStatus() {
        for (ForwarderListener forwarderListener : listenerList) {
            forwarderListener.forwarderDynamicStateChanged(this);
        }

        if(ForwardProxy.allowRemoteConsole){
            ForwarderDynamicStateChangedPayload payload= createDynamicStateChangedPayload();
            payload.printOut();
            Command c = new Command(ForwarderDynamicStateChangedPayload.COMMAND,payload);
            ForwardProxy.addServiceMessage(c);
        }
    }
    public void connectionDataChanged(ClientServiceRequest csr){
        for (ForwarderListener forwarderListener : listenerList) {
            forwarderListener.connectionDataChanged(this, csr);
        }

        if(ForwardProxy.allowRemoteConsole){
            ForwarderDynamicStateChangedPayload payload= createDynamicStateChangedPayload();
            payload.printOut();
            Command c = new Command(Command.UPDATE_SERVICE_REQUEST, csr);
            ForwardProxy.addServiceMessage(c);
        }

    }


    private ForwarderDynamicStateChangedPayload createDynamicStateChangedPayload(){
        ForwarderDynamicStateChangedPayload payload = new ForwarderDynamicStateChangedPayload();
        payload.forwarderName = this.toString();
        payload.activeCount          = new AtomicInteger(activeCount.intValue());
        payload.currentRequestId     = new AtomicInteger(currentRequestId.intValue());
        payload.errorCount           = new AtomicInteger(errorCount.intValue());
        payload.totalBytesFromClient = new AtomicLong(totalBytesFromClient.longValue());
        payload.totalBytesFromEngine = new AtomicLong(totalBytesFromEngine.longValue());
        payload.totalBytesToClient   =  new AtomicLong(totalBytesToClient.longValue());
        payload.totalBytesToEngine   =  new AtomicLong(totalBytesToEngine.longValue());
        return payload;
    }

    public void register(ForwarderListener fl) {
        if (listenerList == null) {
            listenerList = new ArrayList<ForwarderListener>();
        }
        listenerList.add(fl);
    }
}
